package com.kqkj.dyrule.marketing.service;

import com.kqkj.dyrule.marketing.beans.EventBean;
import com.kqkj.dyrule.marketing.beans.EventCombinationCondition;
import com.kqkj.dyrule.marketing.querier.ClickHouseQuerier;
import com.kqkj.dyrule.marketing.querier.HbaseQuerier;
import com.kqkj.dyrule.marketing.querier.StateQuerier;
import com.kqkj.dyrule.marketing.utils.ConfigNames;
import com.kqkj.dyrule.marketing.utils.ConnectionUtils;
import com.kqkj.dyrule.marketing.utils.CrossTimeQueryUtil;
import com.kqkj.dyrule.marketing.utils.EventUtil;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.java.tuple.Tuple2;

import java.io.IOException;
import java.util.Map;

/**
 * 触发型规则
 * 1.做过1，任意事件后做过3，任意事件后做过5            "1.*3.*5"
 * 2.做过1紧跟2，任意事件后做过3                      "12.*3"
 * 3.做过1，2且之间间隔5个任意事件                    "1(\\d{5})2"
 * 4.做过1，2且之间隔了至少一个别的事件                "1(\\d+)2"
 * 5.做过1后紧跟做了3次2，任意事件后做过3              "12{3}.*3"
 * 6.做过1后紧跟做了2-3次2，后紧跟着做过3              "12{2,3}3"
 * 7.做过1，后面做过3，中间不能发生2                   "1(?!2).*3"
 * 8.做过1，后面做过3，中间不能发生2或者4              "1(?![24]).*3"
 * 9.做过1，后面紧跟着做过3或者5，再之后任意事件后做过2   "1(?=[35]).*2"
 */


public class TriggerModelRuleMatchServiceImpl {
    ClickHouseQuerier clickHouseQuerier;
    HbaseQuerier hbaseQuerier;
    StateQuerier stateQuerier;
    public TriggerModelRuleMatchServiceImpl(ListState<EventBean> listState) throws Exception {
        Config load = ConfigFactory.load();
        //hbase连接
        hbaseQuerier = new HbaseQuerier(ConnectionUtils.getHbaseConnection(), load.getString(ConfigNames.HBASE_PROFILE_TABLE), load.getString(ConfigNames.HBASE_PROFILE_FAMILY));
        //CK连接
        clickHouseQuerier = new ClickHouseQuerier(ConnectionUtils.getCkConnection());
        //flink state连接
        stateQuerier = new StateQuerier(listState);
    }

    //用户画像hbase匹配
    public boolean matchProfileCondition(Map<String,String> profileCondition,String deviceId) throws IOException {
         return hbaseQuerier.queryProfileConditionIsMatch(profileCondition, deviceId);
    }



    /**
     * 计算一个行为组合事件是否匹配
     * @param combinationCondition
     * @param event
     * @return
     */
    public boolean matchEventCombinationCondition(EventCombinationCondition combinationCondition, EventBean event) throws Exception {
        //计算划分的标记时间戳
        long segmentPoint = CrossTimeQueryUtil.getSegmentPoint(event.getTimeStamp());
        //判断是否为分界点
        long conditionStart = combinationCondition.getTimeRangeStart();
        long conditionEnd = combinationCondition.getTimeRangeEnd();
        String keyByValue = event.getKeyByValue();
        if (conditionStart >= segmentPoint){
            //查state状态
            int count1 = stateQuerier.queryEventCombinationConditionCount(keyByValue, combinationCondition, conditionStart, conditionEnd);
            return count1 >= combinationCondition.getMinLimit() && count1 <= combinationCondition.getMaxLimit();
        }
        else if (conditionEnd < segmentPoint){
            //查clickhouse
            Tuple2<String,Integer> resTupleCk = clickHouseQuerier.queryEventCombinationConditionCount(keyByValue, combinationCondition, conditionStart, conditionEnd);
            return resTupleCk.f1 >= combinationCondition.getMinLimit() && resTupleCk.f1 <= combinationCondition.getMaxLimit();
        }
        else{
            //跨区查询
            //先查flink，看能否提前结束
            int count3 = stateQuerier.queryEventCombinationConditionCount(keyByValue, combinationCondition, segmentPoint, conditionEnd);
            if (count3 >= combinationCondition.getMinLimit() ) return true;
            //再查clickhouse,看能否提前结束
            Tuple2<String,Integer> resTupleCk = clickHouseQuerier.queryEventCombinationConditionCount(keyByValue, combinationCondition, conditionStart, segmentPoint,true);
            System.out.println("clickhouse查询结果"+resTupleCk);
            if (resTupleCk.f1 >= combinationCondition.getMinLimit()) return true;
            /**
             * 此处调用违反了buffer层级的原则，导致此处会跳过缓存处理
             * 分段组合查询，先从ck查询字符串序列，再从state查询字符串序列，拼接后作为整体匹配正则表达式
             */
            String str2 = stateQuerier.queryEventCombinationConditionStr(keyByValue, combinationCondition, segmentPoint, conditionEnd);
            count3 = EventUtil.sequenceStrMatchRegexGroupCount(resTupleCk.f0 + str2, combinationCondition.getMatchPattern());
            return count3 >= combinationCondition.getMinLimit() && count3 <= combinationCondition.getMaxLimit();
        }
    }
}
